package today.rocky.boot.kafka.client.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

/**
 * @author Rocky
 * @version 1.0
 * @date 2020/3/18 13:05
 */
public class KafkaReceiver extends Thread {
    // sh kafka-topics.sh --create --zookeeper 172.16.113.108:2181 --replication-factor 1 --partitions 1 --topic test
    // sh kafka-topics.sh --list --zookeeper 172.16.113.108:2181

    KafkaConsumer<Integer,String> consumer;
    String topic;

    public static void main(String[] args) {
        //new KafkaReceiver("test").start();
        new KafkaReceiver("partition_test").start();
    }

    public KafkaReceiver(String topic) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.48.131:9092");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG,"rocky-consumer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"rocky-gid");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"30000");
        //自动确认
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        this.consumer = new KafkaConsumer<Integer, String>(properties);
        this.topic = topic;
    }

    @Override
    public void run() {
        //指定消费某个分区
        TopicPartition topicPartition=new TopicPartition(topic,0);
        consumer.assign(Arrays.asList(topicPartition));
        //consumer.subscribe(Collections.singleton(this.topic));
        while (true){
            ConsumerRecords<Integer, String> poll = consumer.poll(Duration.ofSeconds(1));
            poll.forEach(record->{
                System.out.println(record.key()+"->"+record.value()+"->"+record.offset());
            });
        }


    }
}
